package com.fast.bigdata.hourglass.config;

import akka.actor.ActorSystem;
import bigdata.hourglass.common.utils.ZkUtils;
import bigdata.hourglass.utils.AkkaRpcProviderFactory;
import bigdata.hourglass.utils.AkkaUtils;
import bigdata.hourglass.utils.HourglassSystemConstant;
import com.typesafe.config.Config;
import org.I0Itec.zkclient.ZkClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;

/**
 * @author leo.jie (weixiao.me@aliyun.com)
 * @version 1.0
 * @organization CC
 * @website https://www.jlpyyf.com
 * @date 2019/12/8 10:38 下午
 * @since 1.0
 */
@Configuration
public class AkkaConfig {
    @Autowired
    private ApplicationContext applicationContext;
    @Autowired
    private SpringExtension springExtension;
    @Autowired
    private EnvConfig envConfig;

    @Bean
    public ActorSystem actorSystem() {
        ActorSystem actorSystem = ActorSystem.create("HourglassServerSystem", getConfig());
        springExtension.initialize(applicationContext);
        return actorSystem;
    }

    @Bean
    public ZkClient getZkClient() {
        return ZkUtils.getZkClient(envConfig.getProperty("zkServers"));
    }

    @Bean
    public Config getConfig() {
        String port = envConfig.getProperty("engine.port", HourglassSystemConstant.DEFAULT_AKKA_TRANSPORT);
        String logLevel = envConfig.getProperty("akka.logLevel", HourglassSystemConstant.DEFAULT_AKKA_LOG_LEVEL);
        AkkaRpcProviderFactory akkaRpcProviderFactory = new AkkaRpcProviderFactory();
        akkaRpcProviderFactory.initConfig(Integer.parseInt(port), logLevel);
        return akkaRpcProviderFactory.getAkkaConfig(getZkClient());
    }
}
